package com.course.rocketmq.producer;

import com.alibaba.fastjson.JSONObject;
import com.course.model.entity.redisson.UserOrder;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Collection;

/**
 * @PROJECT_NAME: course
 * @DESCRIPTION:
 * @Author: 涂玄武
 * @DATE: 2021/6/8 14:51
 */
@Component
public class ProducerSimple {

    private static final Logger logger = LoggerFactory.getLogger(ProducerSimple.class);

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息
     *
     * @param topic
     * @param msg
     */
    public void sendSyncMsg(String topic, String msg) {
        rocketMQTemplate.syncSend(topic, msg);
    }

    /**
     * 发送异步消息
     *
     * @param topic
     * @param msg
     */
    public void sendASyncMsg(String topic, String msg) {
        rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                //成功回调
                logger.info("成功回调：{}", sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                //异常回调
                logger.info("异常回调：{}", throwable.getMessage());
            }
        });
    }

    /**
     * 发送单向消息
     * @param topic
     * @param msg
     */
    public void sendOneWayMsg(String topic, String msg) {
        this.rocketMQTemplate.sendOneWay(topic,msg);
    }

    /**
     * 消息内容为json格式
     * @param topic
     * @param userOrder
     */
    public void sendMsgByJson(String topic, UserOrder userOrder){
        //发送同步消息，消息内容将userOrder转为json
        this.rocketMQTemplate.convertAndSend(topic,userOrder);
        logger.info("消息内容为json格式:{}",userOrder);
    }

    /**
     * 发送同步延迟消息
     * 消息内容为json格式
     * @param topic
     * @param order
     * @throws JsonProcessingException
     */
    public void sendMsgByJsonDelaySync(String topic,UserOrder order) throws JsonProcessingException {
        //发送同步消息，消息内容将userOrder转为json
        Message<UserOrder> message = MessageBuilder.withPayload(order).build();
        //指定发送超时时间（毫秒）和延迟等级
        this.rocketMQTemplate.syncSend(topic,message,1000,3);
    }

    /**
     * 发送异步延迟消息
     * 消息内容为json格式
     * @param topic
     * @param order
     * @throws Exception
     */
    public void sendMsgByJsonDelayASync(String topic,UserOrder order) throws Exception{
        //发送同步消息，消息内容将userOrder转为json
        String json = JSONObject.toJSONString(order);
        org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message(topic,json.getBytes(StandardCharsets.UTF_8));
        message.setDelayTimeLevel(3);
        this.rocketMQTemplate.getProducer().send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                //成功回调
                logger.info("成功回调：{}", sendResult.getSendStatus());
            }

            @Override
            public void onException(Throwable throwable) {
                //异常回调
                logger.info("异常回调：{}", throwable.getMessage());
            }
        });
    }
}
